package com.pajk.user.service.common.metaq;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.MessageFormat;
import java.util.Date;
import java.util.HashMap;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.common.lang.StringUtil;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.*;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.pajk.user.dal.mapper.CreditHisMapper;
import com.pajk.user.dal.model.CreditHisDO;
import com.pajk.user.service.credit.CreditAddConsumer;
import com.pajk.user.util.HessianUtils;

/**
 * Created by chenhaihua on 14-4-22.
 */
public class TransMsgSender {
    private final static Logger   logger = LoggerFactory.getLogger(TransMsgSender.class);
    private TransactionMQProducer producer;

    private String                nameServer;

    private boolean test = false;

    @Resource
    private CreditHisMapper creditHisMapper;
    
    public void init() throws MQClientException {
        //为本地测试关闭metaq的接收
        if ("localTest".equals(nameServer)) {
            test = true;
            return;
        }
        //设置nameserver 地址
        if (StringUtil.isBlank(System.getProperty("rocketmq.namesrv.domain"))) {
            System.setProperty("rocketmq.namesrv.domain", nameServer);
        }
        TransactionCheckListener transactionCheckListener = new TransactionCheckListener() {
            @Override
            public LocalTransactionState checkLocalTransactionState(MessageExt messageExt) {
                TopicEnum topic = TopicEnum.findTopicEnum(messageExt.getTopic(),
                    messageExt.getTags());
                try {
                    Object msg = HessianUtils.decode(messageExt.getBody());
                    return getCommitStatus(topic, msg, messageExt.getBornTimestamp());
                } catch (IOException e) {
                    logger.error(e.getMessage(), e);
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
            }
        };
        producer = new TransactionMQProducer("P_user_trans");
        try {
            producer.setInstanceName("DEFAULT_TRANS_MSG_SENDER-"+ InetAddress.getLocalHost().getHostName());
        } catch (UnknownHostException e) {
            logger.error("getHostName error",e);
        }
        //事务回查最小并发数
        producer.setCheckThreadPoolMinSize(2);
        //事务回查最大并发数
        producer.setCheckThreadPoolMaxSize(2);
        //队列数
        producer.setCheckRequestHoldMax(2000);
        
        producer.setNamesrvAddr(nameServer);
        
        //设置事务会查监听器
        producer.setTransactionCheckListener(transactionCheckListener);

        producer.start();

        logger.info(MessageFormat.format("事务metaq消息启动成功!nameServer={0},group={1}",
            System.getProperty("rocketmq.namesrv.domain"), "P_user_trans"));
    }

    private LocalTransactionState getCommitStatus(TopicEnum topic, Object msg, long bornTimestamp) {
        switch (topic) {
            case USER_ADD_CREDIT: 
               return getAddCreditStatus(msg);
            default:
                return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

    private LocalTransactionState getAddCreditStatus(Object obj) {
        if (obj instanceof HashMap) {
            try {
                HashMap<String, Object> msg = (HashMap<String, Object>) obj;
                String actionId = (String) msg.get(CreditAddConsumer.MSG_ACTIONID);
                long userId = (Long) msg.get(CreditAddConsumer.MSG_USERID);
                HashMap<String, Object> selectMap = new HashMap<String, Object>();
                selectMap.put("userId", userId);
                selectMap.put("actionId", actionId);
                CreditHisDO creditHisDO = creditHisMapper.getCreditByActionId(selectMap);
                if (creditHisDO != null) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
    
    public boolean sendMessageInTransaction(TopicEnum topicEnum, Object msg,
                                            LocalTransactionExecuter transactionExecuter) {
        logger.info(MessageFormat.format("准备发送事务消息!topic={0},tags={1},message={2}",
            topicEnum.getTopic(), topicEnum.getTags(), msg));
        TransactionSendResult sendResult;
        try {
            Message message = new Message(topicEnum.getTopic(), topicEnum.getTags(),
                HessianUtils.encode(msg));
            //发送事务性消息
            if (test){
                transactionExecuter.executeLocalTransactionBranch(message,null);
                return true;
            }
            sendResult = producer.sendMessageInTransaction(message, transactionExecuter, null);
            if (sendResult.getLocalTransactionState() == LocalTransactionState.ROLLBACK_MESSAGE) {
                logger.warn("发送事务消息失败!topic={0},tags={1},message={2}", topicEnum.getTopic(),
                    topicEnum.getTags(), msg);
                return false;
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            return false;
        }

        return true;
    }

    public void destroy() {
        if (producer != null) {
            producer.shutdown();
            logger.info(MessageFormat.format("事务metaq消息关闭成功!nameServer={0},group={1}",
                System.getProperty("rocketmq.namesrv.domain"), "P_fundselling_transaction"));
        }
    }

    public void setNameServer(String nameServer) {
        this.nameServer = nameServer;
    }

}
